package wsmsgsubpub

import (
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/factory"
	"gitee.com/ymofen/gobase/gotask"
	"gitee.com/ymofen/gobase/subpub"
	"io"
	"sync/atomic"
)

// 将数据分流给NtripClient
type PubWsMsgSession struct {
	closedflag int32
	notify     subpub.ISubPub
	sessionid  string

	conf gobase.StrMap

	copybufflag int8

	operaRef *gobase.ObjectInnerRef
	opera    *WsCliNode
}

func CreatePubWsMsgSession(args ...interface{}) (rval interface{}, err error) {
	plg := &PubWsMsgSession{
		sessionid: args[0].(string),
		conf:      args[1].(gobase.StrMap),
		notify:    args[2].(subpub.ISubPub),
	}
	return plg, nil
}

func (this *PubWsMsgSession) Pub(topic string, max int, args ...interface{}) int {
	ex := this.notify
	if ex == nil {
		return -1
	}

	if len(args) == 0 {
		ex.Pub(this.sessionid, 0, subpub.ACTION_ERROR, fmt.Errorf("没有要Pub的数据"))
		return -1
	}

	opera := this.opera
	if opera == nil {
		return -1
	}

	if buf, ok := args[0].([]byte); ok {
		n, err := opera.Pub(topic, buf)
		if err != nil {
			ex.Pub(this.sessionid, 0, subpub.ACTION_ERROR, err.Error())
			return -1
		}
		return n
	} else if str, ok := args[0].(string); ok {
		n, err := opera.Pub(topic, []byte(str))
		if err != nil {
			ex.Pub(this.sessionid, 0, subpub.ACTION_ERROR, err.Error())
			return -1
		}
		return n
	} else {
		ex.Pub(this.sessionid, 0, subpub.ACTION_ERROR, fmt.Errorf("参数不是[]byte类型"))
		return -1
	}
}

func (this *PubWsMsgSession) PostData(args ...interface{}) error {
	return nil
}

func (this *PubWsMsgSession) onNtripClientEvt(id, topic string, args ...interface{}) (ok bool) {
	ex := this.notify
	if ex == nil {
		return false
	}

	if len(args) < 2 {
		return false
	}
	//action := args[0].(int)

	return false
}

func (this *PubWsMsgSession) Start() {

	connstr := fmt.Sprintf("msgtype=%s&url=%s&waitresp=%d&connrand=%s",
		this.conf.StringByName("msgtype", "binmsg"),
		this.conf.StringByName("url", ""),
		this.conf.Int8ByName("waitresp", 1),
		this.conf.StringByName("connrand", ""))

	this.operaRef = DefaultWsMsgCliGroup.CheckGetRef(connstr)
	this.opera = this.operaRef.GetObject().(*WsCliNode)
	this.opera.ex.Sub(this.sessionid, "evt", func(id, topic string, args ...interface{}) (ok bool) {
		//this.pubfun(this.sessionid, args...)
		return true
	})
	this.opera.checkStart()
}

func (this *PubWsMsgSession) Close() error {
	if !atomic.CompareAndSwapInt32(&this.closedflag, 0, 1) {
		return io.ErrClosedPipe
	}
	if this.operaRef != nil {
		this.opera.ex.Unsub(this.sessionid, "evt")
		this.opera = nil

		DefaultWsMsgCliGroup.ReleaseRef(this.operaRef)
		this.operaRef = nil
	}

	gotask.DefaultPatrolTask().DelTask(this.sessionid)

	this.notify.Pub(this.sessionid, 0, subpub.ACTION_DISCONNECTED)
	this.notify = nil
	return nil
}

func init() {
	factory.RegisterCreateFunc("pub.wsmsg", CreatePubWsMsgSession)
}
